package com.abyss.table;


import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;

/**
 * 演示FlinkSQL的批处理
 */
public class BatchSQLDemo {
    public static void main(String[] args) throws Exception {
        // 1. 注册table的运行环境
        // 2. 完成将dataset注册为table对象的操作
        // 3. 执行table api完成sql查询
        // 4. 将结果转换回dataset

        /*
        1. 准备环境
         */
        // 1.1 批处理环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 1.2 取得批处理table的环境
        BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);

        /*
        2. 将dataset注册为table对象
         */
        // 2.1 准备source(dataset)
        DataSource<Student> studentDataSet = env.fromElements(
                new Student("潇潇", 22, "male"),
                new Student("甜甜", 11, "female"),
                new Student("蛋蛋", 10, "male"),
                new Student("刚刚", 9, "male"),
                new Student("妞妞", 16, "female")
        );
        // 2.2 将dataset注册为table对象
        Table table = tEnv.fromDataSet(studentDataSet, "name, age, sex");

        /*
        3. 执行table api完成sql查询
         */
        // 3.1 按照Table API的形式查询
        // from where group by(聚合) having order by limit select
        Table tableResult1 = table.where("age > 10").groupBy("sex").select("sex, max(age) as maxAge");

        // 3.2 按照SQL的形式查询
        tEnv.createTemporaryView("student", studentDataSet, "name, age, sex");
        Table tableResult2 = tEnv.sqlQuery("SELECT sex, max(age) AS maxAge FROM student WHERE age > 10 GROUP BY sex");

        /*
        4. 将table结果转换回dataset对象
         */
        DataSet<QueryResult> resultDataSet1 = tEnv.toDataSet(tableResult1, QueryResult.class);
        resultDataSet1.print("result1>>>");

        DataSet<QueryResult> resultDataSet2 = tEnv.toDataSet(tableResult2, QueryResult.class);
        resultDataSet2.print("result2>>>");

        env.execute();
    }

    /**
     * 定义查询结果的javabean类
     */
    public static class QueryResult {
        private String sex;
        private int maxAge;

        @Override
        public String toString() {
            return "QueryResult{" +
                    "sex='" + sex + '\'' +
                    ", maxAge=" + maxAge +
                    '}';
        }

        public String getSex() {
            return sex;
        }

        public void setSex(String sex) {
            this.sex = sex;
        }

        public int getMaxAge() {
            return maxAge;
        }

        public void setMaxAge(int maxAge) {
            this.maxAge = maxAge;
        }

        public QueryResult() {
        }

        public QueryResult(String sex, int maxAge) {
            this.sex = sex;
            this.maxAge = maxAge;
        }
    }

    /**
     * Student类的javabean实现
     */
    public static class Student {
        private String name;
        private int age;
        private String sex;

        @Override
        public String toString() {
            return "Student{" +
                    "name='" + name + '\'' +
                    ", age=" + age +
                    ", sex='" + sex + '\'' +
                    '}';
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public int getAge() {
            return age;
        }

        public void setAge(int age) {
            this.age = age;
        }

        public String getSex() {
            return sex;
        }

        public void setSex(String sex) {
            this.sex = sex;
        }

        public Student() {}

        public Student(String name, int age, String sex) {
            this.name = name;
            this.age = age;
            this.sex = sex;
        }
    }
}